package kuoge.practice.hash.consistenthash;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HashSet;
import java.util.Set;
import java.util.TreeMap;

/**
 * Ketama is an implementation of a consistent hashing algorithm, meaning you can add or remove servers from the memcached pool without causing a complete remap of all keys.
 * Here’s how it works:
 * * Take your list of servers (eg: 1.2.3.4:11211, 5.6.7.8:11211, 9.8.7.6:11211)
 * * Hash each server string to several (100-200) unsigned ints
 * * Conceptually, these numbers are placed on a circle called the continuum. (imagine a clock face that goes from 0 to 2^32)
 * * Each number links to the server it was hashed from, so servers appear at several points on the continuum, by each of the numbers they hashed to.
 * * To map a key->server, hash your key to a single unsigned int, and find the next biggest number on the continuum. The server linked to that number is the correct server for that key.
 * * If you hash your key to a value near 2^32 and there are no points on the continuum greater than your hash, return the first server in the continuum.
 * If you then add or remove a server from the list, only a small proportion of keys end up mapping to different servers.
 */
public class KetamaConsistentHash {
    private final Set<Node> nodes = new HashSet<>();
    private final TreeMap<Long, Node>  hashCycle = new TreeMap<>();
    private static final int NODE_COPY_NUM = 10;

    public KetamaConsistentHash() {
        final Node node_1 = new Node("10.11.101.1", true);
        final Node node_2 = new Node("10.11.101.2", true);
        final Node node_3 = new Node("10.11.101.3", true);
        nodes.add(node_1);
        nodes.add(node_2);
        nodes.add(node_3);
    }

    /**
     * 这里(poc)只考虑第一次新建哈希环，不考虑后续节点的变化引起的重建
     *
     * todo: lock
     */
    public void buildHashCycle()  {
        for (final Node node : nodes) {
            // 每4个虚拟节点组成一环
            for (int i = 0;i < NODE_COPY_NUM/4;i++) {
                // 这组虚拟节点唯一的16bit的名称
                final byte[] md5s;
                try {
                    md5s = MessageDigest.getInstance("MD5").digest((node.getIp() + "@" + i).getBytes(StandardCharsets.UTF_8));
                } catch (NoSuchAlgorithmException e) {
                    e.printStackTrace();
                    return;
                }
                // 在这16bit的md5串中，每4个为一个long，作为虚拟节点的key
                for (int offset = 0;offset < 4;offset++) {
                    long hash = getHash(md5s, offset);
                    // 记录虚拟节点的hash值和节点的关联关系
                    hashCycle.put(hash, node);
                }
            }

        }
    }

    /**
     *
     * todo: lock
     */
    public Node findNodeByKey(final String key) {
        try {
            final long keyHash = getHash(MessageDigest.getInstance("MD5").digest((key).getBytes(StandardCharsets.UTF_8)), 0);
            if (hashCycle.containsKey(keyHash)) {
                return hashCycle.get(keyHash);
            }else {
                final Long higherKey = hashCycle.higherKey(keyHash);
                System.out.println("higherKeyHash:" + higherKey);
                System.out.println("keyHash:" + keyHash);
                return hashCycle.get(higherKey);
            }
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
            return null;
        }
   }

    public TreeMap<Long, Node> getHashCycle() {
        return hashCycle;
    }

    private long getHash(byte[] md5s, int offset) {
        return (((long) (md5s[3 + offset * 4] & 0xFF) << 24)
                | ((long) (md5s[2 + offset * 4] & 0xFF) << 16)
                | ((long) (md5s[1 + offset * 4] & 0xFF) << 8)
                | ((long) md5s[offset * 4] & 0xFF)) & 0xffffffffL;
    }

    public static void main(String[] args) {
        final KetamaConsistentHash ketamaConsistentHash = new KetamaConsistentHash();
        ketamaConsistentHash.buildHashCycle();
        System.out.println(ketamaConsistentHash.getHashCycle().size());
        System.out.println(ketamaConsistentHash.getHashCycle().descendingKeySet());
        final String uuid = "4c20c9304a164df7a5b1aa371aa917bb";
        final Node node = ketamaConsistentHash.findNodeByKey(uuid);
        System.out.printf(node.getIp());
    }

}
